﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using SIEGate.Core.Adapters;

namespace SIEGate.Core.MessageBusses
{
    public class RoutingTablesMessageBus : IMessageBus
    {
        private class Route
        {
            public RoutingTablesMessageBusProxy Producer;
            public ISet<RoutingTablesMessageBusProxy> Consumers;
            public object SyncRoot;

            public Route()
            {
                Consumers = new HashSet<RoutingTablesMessageBusProxy>();
                SyncRoot = new object();
            }
        }

        private ConcurrentDictionary<Guid, Route> m_routes; 

        public RoutingTablesMessageBus()
        {
            m_routes = new ConcurrentDictionary<Guid, Route>();
        }

        public IMessageBusProxy GetProxy(string name = "none")
        {
            return new RoutingTablesMessageBusProxy(this);
        }

        public void Stats()
        {
            // Not implemented.
        }

        public void EnablePath(Guid messageId, OutputAdapter adapter)
        {
            // Not even used.
        }

        public void DisablePath(Guid messageId, OutputAdapter adapter)
        {
            // Not even used.
        }

        internal void RegisterProducer(Guid streamID, RoutingTablesMessageBusProxy proxy)
        {
            Route route = m_routes.GetOrAdd(streamID, guid => new Route());

            lock (route.SyncRoot)
            {
                route.Producer = proxy;

                if (route.Consumers.Count > 0)
                    route.Producer.OnStartSubscription(streamID);
            }
        }

        internal void RegisterConsumer(Guid streamID, RoutingTablesMessageBusProxy proxy)
        {
            Route route = m_routes.GetOrAdd(streamID, guid => new Route());

            lock (route.SyncRoot)
            {
                if ((object)route.Producer != null && route.Consumers.Count == 0)
                    route.Producer.OnStartSubscription(streamID);

                route.Consumers.Add(proxy);
            }
        }

        internal void UnregisterProducer(Guid streamID, RoutingTablesMessageBusProxy proxy)
        {
            Route route = m_routes.GetOrAdd(streamID, guid => new Route());

            lock (route.SyncRoot)
            {
                if (proxy == route.Producer)
                    route.Producer = null;

                proxy.OnEndSubscription(streamID);
            }
        }

        internal void UnregisterConsumer(Guid streamID, RoutingTablesMessageBusProxy proxy)
        {
            Route route = m_routes.GetOrAdd(streamID, guid => new Route());

            lock (route.SyncRoot)
            {
                route.Consumers.Remove(proxy);

                if ((object)route.Producer != null && route.Consumers.Count == 0)
                    route.Producer.OnEndSubscription(streamID);
            }
        }

        internal void RouteMeasurements(RoutingTablesMessageBusProxy sender, params IMeasurement[] measurements)
        {
            Route route;

            foreach (IMeasurement measurement in measurements)
            {
                if (!m_routes.TryGetValue(measurement.StreamID, out route))
                    continue;

                lock (route.SyncRoot)
                {
                    if (route.Producer == sender)
                    {
                        foreach (RoutingTablesMessageBusProxy proxy in route.Consumers)
                            proxy.Enqueue(measurement);
                    }
                }
            }
        }
    }
}
